package Yanxin.atguigu.yx.app.App_03_DwdApp.Dwd_01_Traffic;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import Yanxin.atguigu.yx.app.APP_01_BaseApp.BaseApp;
import Yanxin.atguigu.yx.util.AtguiguUtil;
import Yanxin.atguigu.yx.util.FlinkSinkUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import Yanxin.atguigu.yx.common.Constant;

import java.util.HashMap;

public class Dwd_00_Traffic_splitFlow_App extends BaseApp {

    private final String PAGE = "page";
    private final String APPVIDEO = "appVideo";
    private final String ERR = "err";
    private final String ACTION = "action";
    private final String DISPLAY = "display";
    private final String START = "start";


    public static void main(String[] args) {
         new Dwd_00_Traffic_splitFlow_App().init(
                3001,
                2,
                "Dwd_00_Traffic_splitFlow_App",
                Constant.TOPIC_ODS_LOG);

    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {

        //1.etl
        SingleOutputStreamOperator<JSONObject> etlStream = etl(stream);


        //2.程序启动的时候，纠正老用户出现is_new=1变为0，在变成1的情况
        /**
         * 用户第一次访问埋入神策 SDK 页面的当天（即第一天），
         * JS SDK 会在网页的 cookie 中设置一个首日访问的标记，并设置第一天 24 点之前，该标记为 true，即第一天触发的网页端所有事件中，is_new = 1。
         * 第一天之后，该标记则为 false，即第一天之后触发的网页端所有事件中，is_new = 0；
         */
        SingleOutputStreamOperator<JSONObject> validatedStream = validateNewOrOld(etlStream);

        //3.分流
        HashMap<String, DataStream<JSONObject>> resultStream = splitStream(validatedStream);

        //4.将分流结果传到kafka
        writeToKafka(resultStream);

    }

    private void writeToKafka(HashMap<String, DataStream<JSONObject>> resultStream) {

        //将START主流传进kafka
        resultStream.get(START) //通过hashmap的key拿到主流start
//                .map(JSONObject-> JSONObject.toString()) //将JSONObject对象转成String字符串才能传进kafka
                .map(JSONObject-> JSONObject.toString())
                .addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_START));


        resultStream.get(ACTION).map(JSONObject-> JSONObject.toString()).addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ACTION));
        resultStream.get(DISPLAY).map(JSONObject-> JSONObject.toString()).addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_DISPLAY));
        resultStream.get(ERR).map(JSONObject-> JSONObject.toString()).addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_ERR));
        resultStream.get(PAGE).map(JSONObject-> JSONObject.toString()).addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_PAGE));

        resultStream.get(APPVIDEO).map(JSONObject-> JSONObject.toString()).addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWD_TRAFFIC_APPVIDEO));

    }

    private HashMap<String, DataStream<JSONObject>> splitStream(SingleOutputStreamOperator<JSONObject> validatedStream) {

        OutputTag<JSONObject> displayTag = new OutputTag<JSONObject>("display") {};
        OutputTag<JSONObject> actionTag = new OutputTag<JSONObject>("action") {};
        OutputTag<JSONObject> errTag = new OutputTag<JSONObject>("err") {};
        OutputTag<JSONObject> pageTag = new OutputTag<JSONObject>("page") {};
        OutputTag<JSONObject> appVideoTag = new OutputTag<JSONObject>("appVideo") {};

        SingleOutputStreamOperator<JSONObject> finalSplitStream = validatedStream
                .process(new ProcessFunction<JSONObject, JSONObject>() {
                    @Override
                    public void processElement(JSONObject value,
                                               Context ctx,
                                               Collector<JSONObject> out) throws Exception {

                        //分流，1.判断是否是启动日志
                        if (value.containsKey("start")) {
                            out.collect(value);

                            //分流，2.不是启动日志，剩余的日志有四个
                        } else {
                            //提取通用json的common
                            JSONObject common = value.getJSONObject("common");
                            JSONObject page = value.getJSONObject("page");
                            Long ts = value.getLong("ts");


                            //2.0 由于appVideo有数据带有err，需要从中删除，下面拿err的数据
                            /**
                             * {
                             *   "appVideo": {
                             *     "play_sec": 29,
                             *     "position_sec": 30,
                             *     "video_id": "4680"
                             *   },
                             *   "common": {
                             *     "ar": "12",
                             *     "ba": "Xiaomi",
                             *     "ch": "xiaomi",
                             *     "is_new": "1",
                             *     "md": "Xiaomi 9",
                             *     "mid": "mid_132",
                             *     "os": "Android 11.0",
                             *     "sc": "1",
                             *     "sid": "b2886503-a7f1-4a25-a253-b4e03a954f32",
                             *     "uid": "95",
                             *     "vc": "v2.1.134"
                             *   },
                             *   "err": {
                             *     "error_code": 2785,
                             *     "msg": " Exception in thread \\  java.net.SocketTimeoutException\\n \\tat com.atguigu.edu2021.mock.log.AppError.main(AppError.java:xxxxxx)"
                             *   },
                             *   "ts": 1661915890663
                             * }
                             */
                            if (value.containsKey("appVideo")){
                                ctx.output(appVideoTag, value);
                                value.remove("appVideo");
                            }


                            //2.1 displays曝光displays
                            /**
                             *  "displays": [
                             *     {
                             *       "display_type": "activity",
                             *       "item": "1",
                             *       "item_type": "activity_id",
                             *       "order": 1,
                             *       "pos_id": 2
                             *     },
                             *     {
                             *       "display_type": "query",
                             *       "item": "32",
                             *       "item_type": "sku_id",
                             *       "order": 2,
                             *       "pos_id": 2
                             *     },
                             *     {
                             *       "display_type": "query",
                             *       "item": "34",
                             *       "item_type": "sku_id",
                             *       "order": 3,
                             *       "pos_id": 2
                             *     }
                             *   ],
                             */
                            //displays为一个json数组，
                            JSONArray displays = value.getJSONArray("displays");
                            if (displays != null) {
                                //遍历数组内的每一个内部json
                                for (int i = 0; i < displays.size(); i++) {
                                    JSONObject display = displays.getJSONObject(i);
                                    //每个display都加上对应的common以及page信息,并且额外增加一个key->ts
                                    display.putAll(common);
                                    display.putAll(page);
                                    display.put("ts", ts);
                                    //将其作为侧输出流输出
                                    ctx.output(displayTag, display);
                                }
                                //已经处理完display的内容，此时方法的输入数据JSONObject value要删除displays这个key，否则影响下面对外输出的结果
                                value.remove("displays");
                            }

                            //2.2 活动actions 也是一个json数组
                            JSONArray actions = value.getJSONArray("actions");
                            if (actions != null) {
                                //遍历数组内的每一个内部json
                                for (int i = 0; i < actions.size(); i++) {
                                    JSONObject action = actions.getJSONObject(i);
                                    //每个action都加上对应的common以及page信息
                                    action.putAll(common);
                                    action.putAll(page);
                                    //将其作为侧输出流输出
                                    ctx.output(actionTag, action);
                                }
                                //已经处理完action的内容，此时方法的输入数据JSONObject value要删除actions这个key，否则影响下面对外输出的结果
                                value.remove("actions");
                            }

                            //2.3 处理错误
                            if (value.containsKey("err")) {
                                //有些err里面包含了displays 前面已经删除，不用担心侧输出流会再输出一次全部的value
                                ctx.output(errTag, value);
                                value.remove("err");

                            }

                            //2.4 处理page
                            if (page != null) {
                                //有些err里面包含了displays 前面已经删除，不用担心侧输出流会再输出一次全部的value
                                ctx.output(pageTag, value);
                            }


                        }

                    }
                });
//        System.out.println("我在分流");
//        finalSplitStream.print("主流");
//        finalSplitStream.getSideOutput(displayTag).print("display");
//        finalSplitStream.getSideOutput(actionTag).print("action");
//        finalSplitStream.getSideOutput(errTag).print("err");
//        finalSplitStream.getSideOutput(pageTag).print("page");


        HashMap<String, DataStream<JSONObject>> result = new HashMap<>();

        result.put(START,finalSplitStream);
        result.put(DISPLAY,finalSplitStream.getSideOutput(displayTag));
        result.put(ACTION,finalSplitStream.getSideOutput(actionTag));
        result.put(ERR,finalSplitStream.getSideOutput(errTag));
        result.put(PAGE,finalSplitStream.getSideOutput(pageTag));
        result.put(APPVIDEO,finalSplitStream.getSideOutput(appVideoTag));

        return result;


    }

    private SingleOutputStreamOperator<JSONObject> validateNewOrOld(SingleOutputStreamOperator<JSONObject> etlStream) {

        //根据etlStream的common中的数据，将is_new=0/1的行为与mid绑定，根据mid对某个电商app的访问行为当成是否是首日访问
        return etlStream
                .keyBy(json -> json.getJSONObject("common").getString("mid"))
                .map(new RichMapFunction<JSONObject, JSONObject>() {

                    private ValueState<String> firstDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //开启一个单值状态，记录日期，方便提取数据的时间戳进行比较
                        firstDateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("firstDateState", String.class));
                    }

                    @Override
                    public JSONObject map(JSONObject value) throws Exception {

//                        System.out.println();
//                        System.out.println("---------------------------");
//                        System.out.println(value);
//                        System.out.println("---------------------------");
//                        System.out.println();

                        /**
                         * 每次进来一个
                         * ---------------------------
                         * {
                         * "common":{"ar":"310000","uid":"657","os":"iOS 13.3.1","ch":"Appstore","is_new":"0","md":"iPhone X","mid":"mid_217497",
                         * "vc":"v2.1.134","ba":"iPhone"},"page":{"page_id":"trade","item":"21,26,27","during_time":1147,
                         * "item_type":"sku_ids",     "last_page_id":"orders_unpaid"},"ts":1655201267000
                         * }
                         * ---------------------------
                         *
                         *  16号   in->is_new
                         *  a in=1  16号作为first     --> 状态第一次为null，更新一下，first=16
                         *  b in=0  16-1号作为first   --> 状态第一次为null，更新一下，first=16-1=15
                         *
                         *  17号
                         *  a in=1  16号作为first    --> 错误，16与17号不等 in修改为0
                         *  b in=0  16-1号作为first  --> 状态为15，且17号获取到的in=0 正确不用处理
                         *  c in=1  17号作为first    --> 状态第一次为null
                         *
                         *  18号
                         *  a in=1  16号作为first    --> 错误，状态16与18号不等，in修改为0
                         *  b in=1  16-1号作为first  --> 错误，状态15与18号不等，in修改为0
                         *  c in=1  17号作为first    --> 错误，状态17与18号不等，in修改为0
                         */
                        //获取日期
                        Long ts = value.getLongValue("ts");
                        String today = AtguiguUtil.toDate(ts);
                        String yesterday = AtguiguUtil.toDate(ts - (24 * 60 * 60 * 1000));

                        //获取新老用户状态is_new
                        String isNew = value.getJSONObject("common").getString("is_new");

                        //数据如果是首次进来，状态必然为空
                        if (firstDateState.value() == null) {
                            //数据首次进入,先判断isNew值
                            if ("1".equals(isNew)) {
                                firstDateState.update(today);
                            } else if ("0".equals(isNew)) {
                                firstDateState.update(yesterday);
                            }

                            //如果状态firstDateState不为空，说明数据不是首次进来，
                        } else if (firstDateState.value() != null) {

                            //非首次进入的数据，要非常注意is_new==1的情况，
                            // 可能存在后续数据进来，该设备已经是非首日了，但是收到的数据存在is_new==1
                            if ("1".equals(isNew)) {
                                //如果带进来的数据日期不等于首日，说明该数据有问题
                                if (!firstDateState.value().equals(today)) {
                                    System.out.println("纠正错误的is_new标记");
                                    value.getJSONObject("common").put("is_new", "0");
                                }
                            }


                        }
                        return value;
                    }
                });
//                .print();

    }

    private SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> stream) {

        //只要json数据
        return stream
                .filter(jsonString -> {
                    if ("".equals(jsonString) || jsonString.length()<3) {
                        System.out.println(jsonString + "jsonString是一个空字符串，JSON.parseObject(jsonString)会将其转化成null");
                        return false;
                    }

                    //判断时间
                    try {
                        System.out.println(jsonString);

                        JSONObject jsonObject = JSON.parseObject(jsonString);

                        Long ts = jsonObject.getLong("ts");
                        String today = AtguiguUtil.toDate(ts);
                        String now = AtguiguUtil.toDate(System.currentTimeMillis());
                        if (!now.equals(today)){
                            System.out.println("日期有问题");
                            System.out.println(now);
                            System.out.println(today);
                            return false;
                        }

                    } catch (Exception e) {
                        System.out.println("json格式有误，请检查：" + jsonString);
                        e.printStackTrace();
                        return false;
                    }
                    return true;
                })
//                .map(json->{return JSON.parseObject(json);})
                .map(JSON::parseObject);
//                .print();


    }
}
